#
#    Copyright (c) 2023 Project CHIP Authors
#    All rights reserved.
#
#    Licensed under the Apache License, Version 2.0 (the "License");
#    you may not use this file except in compliance with the License.
#    You may obtain a copy of the License at
#
#        http://www.apache.org/licenses/LICENSE-2.0
#
#    Unless required by applicable law or agreed to in writing, software
#    distributed under the License is distributed on an "AS IS" BASIS,
#    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
#    See the License for the specific language governing permissions and
#    limitations under the License.
#

import importlib
import importlib.resources as pkg_resources
import logging
import os
import re
import typing
import xml.etree.ElementTree as ElementTree
import zipfile
from copy import deepcopy
from dataclasses import dataclass, field
from enum import Enum, StrEnum, auto
from importlib.abc import Traversable
from typing import Callable, Optional, Union

import matter.clusters as Clusters
import matter.testing.conformance as conformance_support
from matter.testing.conformance import (OPTIONAL_CONFORM, TOP_LEVEL_CONFORMANCE_TAGS, ConformanceException,
                                        ConformanceParseParameters, feature, is_disallowed, mandatory, optional, or_operation,
                                        parse_callable_from_xml)
from matter.testing.global_attribute_ids import GlobalAttributeIds
from matter.testing.problem_notices import (AttributePathLocation, ClusterPathLocation, CommandPathLocation, DeviceTypePathLocation,
                                            EventPathLocation, FeaturePathLocation, NamespacePathLocation, ProblemNotice,
                                            ProblemSeverity, UnknownProblemLocation)
from matter.tlv import uint

LOGGER = logging.getLogger(__name__)

# Type alias maintained for constants access; actual values are ints at runtime
ACCESS_CONTROL_PRIVILEGE_ENUM = Clusters.AccessControl.Enums.AccessControlEntryPrivilegeEnum

_PRIVILEGE_STR = {
    None: "N/A",
    ACCESS_CONTROL_PRIVILEGE_ENUM.kView: "V",
    ACCESS_CONTROL_PRIVILEGE_ENUM.kOperate: "O",
    ACCESS_CONTROL_PRIVILEGE_ENUM.kManage: "M",
    ACCESS_CONTROL_PRIVILEGE_ENUM.kAdminister: "A",
}


def to_access_code(privilege: int) -> str:
    return _PRIVILEGE_STR.get(privilege, "")


def get_access_privilege_or_unknown(access_value: Optional[int]) -> int:
    """
    Returns the given access_value if not None, otherwise returns the default unknown privilege.
    """
    if access_value is not None:
        return access_value
    return ACCESS_CONTROL_PRIVILEGE_ENUM.kUnknownEnumValue


class SpecParsingException(Exception):
    pass


# passing in feature map, attribute list, command list
ConformanceCallable = conformance_support.Conformance


class DataTypeEnum(StrEnum):
    # String values must match XML tag names exactly for XML parsing (see line 569: container.iter(str(data_type)))
    # Using StrEnum instead of enum with auto() ensures values are predictable strings, not auto-generated integers
    kStruct = 'struct'
    kEnum = 'enum'
    kBitmap = 'bitmap'


@dataclass
class XmlDataTypeComponent:
    value: uint
    name: str
    conformance: ConformanceCallable
    # Additional datatype component fields from cluster XML's
    summary: Optional[str] = None  # For descriptions/documentation
    type_info: Optional[str] = None  # Data type for struct fields
    is_optional: bool = False  # Whether field is optional
    is_nullable: bool = False  # Whether field can be null
    constraints: Optional[dict] = None  # For min/max values, lists, etc.


@dataclass
class XmlDataType:
    # data_type field is kept for validation, testing, and type safety even though objects
    # are stored in separate structs/enums/bitmaps collections. This allows:
    # 1. Test assertions to verify correct parsing (see TestSpecParsingDataType.py)
    # 2. Type checking when objects are passed without collection context
    data_type: DataTypeEnum
    name: str
    components: dict[uint, XmlDataTypeComponent]
    # if this is None, this is a global struct
    cluster_ids: Optional[list[uint]]


@dataclass
class XmlFeature:
    code: str
    name: str
    conformance: ConformanceCallable

    def __str__(self):
        return f'{self.code}: {self.name} conformance {str(self.conformance)}'


@dataclass
class XmlAttribute:
    name: str
    datatype: str
    conformance: ConformanceCallable
    read_access: int
    write_access: int
    write_optional: bool

    def access_string(self):
        read_marker = "R" if self.read_access is not ACCESS_CONTROL_PRIVILEGE_ENUM.kUnknownEnumValue else ""
        write_marker = "W" if self.write_access is not ACCESS_CONTROL_PRIVILEGE_ENUM.kUnknownEnumValue else ""
        read_access_marker = f'{to_access_code(self.read_access)}'
        write_access_marker = f'{to_access_code(self.write_access)}'
        return f'{read_marker}{write_marker} {read_access_marker}{write_access_marker}'

    def __str__(self):
        return f'{self.name}: datatype: {self.datatype} conformance: {str(self.conformance)}, access = {self.access_string()}'


@dataclass
class XmlCommand:
    id: int
    name: str
    conformance: ConformanceCallable
    privilege: int

    def __str__(self):
        return f'{self.name} id:0x{self.id:02X} {self.id} conformance: {str(self.conformance)} privilege: {str(self.privilege)}'


@dataclass
class XmlEvent:
    name: str
    conformance: ConformanceCallable


@dataclass
class XmlCluster:
    name: str
    revision: int
    derived: Optional[str]
    feature_map: dict[str, uint]
    attribute_map: dict[str, uint]
    command_map: dict[str, uint]
    # mask to XmlFeature
    features: dict[uint, XmlFeature]
    # IDs to class
    attributes: dict[uint, XmlAttribute]
    accepted_commands: dict[uint, XmlCommand]
    generated_commands: dict[uint, XmlCommand]
    unknown_commands: list[XmlCommand]
    events: dict[uint, XmlEvent]
    structs: dict[str, XmlDataType]
    enums: dict[str, XmlDataType]
    bitmaps: dict[str, XmlDataType]
    pics: str
    is_provisional: bool


class ClusterSide(Enum):
    SERVER = auto()
    CLIENT = auto()


@dataclass
class XmlDeviceTypeClusterRequirements:
    name: str
    side: ClusterSide
    conformance: ConformanceCallable
    # Feature mask (1 << feature_id) to conformance
    feature_overrides: dict[uint, Callable] = field(default_factory=dict)
    attribute_overrides: dict[uint, Callable] = field(default_factory=dict)
    command_overrides: dict[uint, Callable] = field(default_factory=dict)

    def __str__(self):
        return f'{self.name}: {str(self.conformance)}'


@dataclass
class XmlNamespace:
    """Represents a namespace definition from XML"""
    id: int = 0
    name: str = ""
    tags: dict[int, 'XmlTag'] = field(default_factory=dict)

    def __str__(self) -> str:
        tags_str = '\n  '.join(f"{tag_id:04X}: {tag.name}"
                               for tag_id, tag in sorted(self.tags.items()))
        return f"Namespace 0x{self.id:04X} ({self.name})\n  {tags_str}"


@dataclass
class XmlTag:
    """Represents a tag within a namespace"""
    id: int = 0
    name: str = ""
    description: Optional[str] = None

    def __str__(self) -> str:
        desc = f" - {self.description}" if self.description else ""
        return f"{self.name}{desc}"


@dataclass
class XmlDeviceType:
    name: str
    revision: int
    server_clusters: dict[uint, XmlDeviceTypeClusterRequirements]
    client_clusters: dict[uint, XmlDeviceTypeClusterRequirements]
    # Keeping these as strings for now because the exact definitions are being discussed in DMTT
    classification_class: str
    classification_scope: str
    superset_of_device_type_name: Optional[str] = None
    superset_of_device_type_id: int = 0

    def __str__(self):
        msg = f'{self.name} - Revision {self.revision}, Class {self.classification_class}, Scope {self.classification_scope}\n'
        if self.superset_of_device_type_name:
            msg += f'superset of {self.superset_of_device_type_name} ({self.superset_of_device_type_id})'
        msg += '    Server clusters\n'
        for id, c in self.server_clusters.items():
            msg = msg + f'      {id}: {str(c)}\n'
        msg += '    Client clusters\n'
        for id, c in self.client_clusters.items():
            msg = msg + f'      {id}: {str(c)}\n'
        return msg


class CommandType(Enum):
    ACCEPTED = auto()
    GENERATED = auto()
    # This will happen for derived clusters, where the direction isn't noted. On normal clusters, this is a problem.
    UNKNOWN = auto()


# workaround for aliased clusters PICS not appearing in the xml. Remove this once https://github.com/csa-data-model/projects/issues/461 is addressed
ALIAS_PICS = {0x040C: 'CMOCONC',
              0x040D: 'CDOCONC',
              0x0413: 'NDOCONC',
              0x0415: 'OZCONC',
              0x042A: 'PMICONC',
              0x042B: 'FLDCONC',
              0x042C: 'PMHCONC',
              0x042D: 'PMKCONC',
              0x042E: 'TVOCCONC',
              0x042F: 'RNCONC',
              0x0071: 'HEPAFREMON',
              0x0072: 'ACFREMON',
              0x0405: 'RH',
              0x001C: 'PWM'}

CLUSTER_NAME_FIXES = {0x0036: 'WiFi Network Diagnostics', 0x042a: 'PM25 Concentration Measurement', 0x0006: 'On/Off'}
DEVICE_TYPE_NAME_FIXES = {0x010b: 'Dimmable Plug-In Unit', 0x010a: 'On/Off Plug-in Unit'}


# fuzzy match to name because some of the old specs weren't careful here
def _fuzzy_name(to_fuzz: str):
    to_fuzz = re.sub(r"\(.*?\)|\[.*?\]", "", to_fuzz)
    return to_fuzz.lower().strip().replace(' ', '').replace('/', '')


def get_location_from_element(element: ElementTree.Element, cluster_id: Optional[int]):
    if cluster_id is None:
        cluster_id = 0
    cluster_location = ClusterPathLocation(endpoint_id=0, cluster_id=cluster_id)
    try:
        if element.tag == 'feature':
            return FeaturePathLocation(endpoint_id=0, cluster_id=cluster_id, feature_code=element.attrib['code'])
        if element.tag == 'command':
            return CommandPathLocation(endpoint_id=0, cluster_id=cluster_id, command_id=int(element.attrib['id'], 0))
        if element.tag == 'attribute':
            return AttributePathLocation(endpoint_id=0, cluster_id=cluster_id, attribute_id=int(element.attrib['id'], 0))
        if element.tag == 'event':
            return EventPathLocation(endpoint_id=0, cluster_id=cluster_id, event_id=int(element.attrib['id'], 0))
        return cluster_location
    except (KeyError, ValueError):
        # If we can't find the id or can't parse it
        return cluster_location


def get_conformance(element: ElementTree.Element, cluster_id: Optional[uint]) -> tuple[ElementTree.Element, typing.Optional[ProblemNotice]]:
    for sub in element:
        if sub.tag in TOP_LEVEL_CONFORMANCE_TAGS:
            return sub, None
    location = get_location_from_element(element, cluster_id)
    problem = ProblemNotice(test_name='Spec XML parsing', location=location,
                            severity=ProblemSeverity.WARNING, problem='Unable to find conformance element')
    return ElementTree.Element(OPTIONAL_CONFORM), problem


# Tuple of the root element, the conformance xml element within the root and the optional access element within the root
XmlElementDescriptor = tuple[ElementTree.Element, ElementTree.Element, Optional[ElementTree.Element]]


class ClusterParser:
    # Cluster ID is optional to support base clusters that have no ID of their own.
    def __init__(self, cluster: ElementTree.Element, cluster_id: Optional[uint], name: str):
        self._problems: list[ProblemNotice] = []
        self._cluster = cluster
        self._cluster_id = cluster_id
        self._name = name

        self._derived = None
        self._is_provisional = False
        try:
            classification = next(cluster.iter('classification'))
            hierarchy = classification.attrib['hierarchy']
            if hierarchy.lower() == 'derived':
                self._derived = classification.attrib['baseCluster']
        except (KeyError, StopIteration):
            self._derived = None

        for ids in cluster.iter('clusterIds'):
            for id in ids.iter('clusterId'):
                if id.attrib['name'] == name and list(id.iter('provisionalConform')):
                    self._is_provisional = True

        self._pics: Optional[str] = None
        try:
            classification = next(cluster.iter('classification'))
            self._pics = classification.attrib['picsCode']
        except (KeyError, StopIteration):
            self._pics = None

        if self._cluster_id is not None and int(self._cluster_id) in ALIAS_PICS:
            self._pics = ALIAS_PICS[int(self._cluster_id)]

        self.feature_elements = self.get_all_feature_elements()
        self.attribute_elements = self.get_all_attribute_elements()
        self.command_elements = self.get_all_command_elements()
        self.event_elements = self.get_all_event_elements()
        self.params = ConformanceParseParameters(feature_map=self.create_feature_map(), attribute_map=self.create_attribute_map(),
                                                 command_map=self.create_command_map())

    def get_conformance(self, element: ElementTree.Element) -> ElementTree.Element:
        element, problem = get_conformance(element, self._cluster_id)
        if problem:
            self._problems.append(problem)
        return element

    def get_access(self, element: ElementTree.Element) -> Optional[ElementTree.Element]:
        for sub in element:
            if sub.tag == 'access':
                return sub
        return None

    def get_all_type(self, type_container: str, type_name: str, key_name: str) -> list[XmlElementDescriptor]:
        ret = []
        container_tags = self._cluster.iter(type_container)
        for container in container_tags:
            elements = container.iter(type_name)
            for element in elements:
                try:
                    element.attrib[key_name]
                except KeyError:
                    # This is a conformance tag, which uses the same name
                    continue
                conformance = self.get_conformance(element)
                access = self.get_access(element)
                ret.append((element, conformance, access))
        return ret

    def get_all_feature_elements(self) -> list[XmlElementDescriptor]:
        ''' Returns a list of features and their conformances'''
        return self.get_all_type('features', 'feature', 'code')

    def get_all_attribute_elements(self) -> list[XmlElementDescriptor]:
        ''' Returns a list of attributes and their conformances'''
        return self.get_all_type('attributes', 'attribute', 'id')

    def get_all_command_elements(self) -> list[XmlElementDescriptor]:
        ''' Returns a list of commands and their conformances '''
        return self.get_all_type('commands', 'command', 'id')

    def get_all_event_elements(self) -> list[XmlElementDescriptor]:
        ''' Returns a list of events and their conformances'''
        return self.get_all_type('events', 'event', 'id')

    def create_feature_map(self) -> dict[str, uint]:
        features = {}
        for element, _, _ in self.feature_elements:
            features[element.attrib['code']] = uint(1 << int(element.attrib['bit'], 0))
        return features

    def create_attribute_map(self) -> dict[str, uint]:
        attributes = {}
        for element, conformance, _ in self.attribute_elements:
            attributes[element.attrib['name']] = uint(int(element.attrib['id'], 0))
        return attributes

    def create_command_map(self) -> dict[str, uint]:
        commands = {}
        for element, _, _ in self.command_elements:
            commands[element.attrib['name']] = uint(int(element.attrib['id'], 0))
        return commands

    def parse_conformance(self, conformance_xml: ElementTree.Element) -> Optional[ConformanceCallable]:
        try:
            return parse_callable_from_xml(conformance_xml, self.params)
        except ConformanceException as ex:
            # Just point to the general cluster, because something is mismatched, but it's not clear what
            location = ClusterPathLocation(endpoint_id=0, cluster_id=self._cluster_id if self._cluster_id is not None else 0)
            self._problems.append(ProblemNotice(test_name='Spec XML parsing', location=location,
                                                severity=ProblemSeverity.WARNING, problem=str(ex)))
            return None

    def parse_write_optional(self, element_xml: ElementTree.Element, access_xml: Optional[ElementTree.Element]) -> bool:
        if access_xml is None:
            return False
        return access_xml.attrib['write'] == 'optional'

    def parse_access(self, element_xml: ElementTree.Element, access_xml: Optional[ElementTree.Element], conformance: ConformanceCallable) -> tuple[Optional[int], Optional[int], Optional[int]]:
        ''' Returns a tuple of access types for read / write / invoke'''
        def str_to_access_type(privilege_str: str) -> int:
            if privilege_str == 'view':
                return ACCESS_CONTROL_PRIVILEGE_ENUM.kView
            if privilege_str == 'operate':
                return ACCESS_CONTROL_PRIVILEGE_ENUM.kOperate
            if privilege_str == 'manage':
                return ACCESS_CONTROL_PRIVILEGE_ENUM.kManage
            if privilege_str == 'admin':
                return ACCESS_CONTROL_PRIVILEGE_ENUM.kAdminister
            # We don't know what this means, for now, assume no access and mark a warning

            location = get_location_from_element(element_xml, self._cluster_id)
            self._problems.append(ProblemNotice(test_name='Spec XML parsing', location=location,
                                                severity=ProblemSeverity.WARNING, problem=f'Unknown access type {privilege_str}'))
            return ACCESS_CONTROL_PRIVILEGE_ENUM.kUnknownEnumValue

        if access_xml is None:
            # Derived clusters can inherit their access from the base and that's fine, so don't add an error
            # Similarly, pure base clusters can have the access defined in the derived clusters. If neither has it defined,
            # we will determine this at the end when we put these together.
            # Things with deprecated conformance don't get an access element, and that is also fine.
            # If a device properly passes the conformance test, such elements are guaranteed not to appear on the device.

            if self._derived is not None or is_disallowed(conformance):
                return (None, None, None)

            location = get_location_from_element(element_xml, self._cluster_id)
            self._problems.append(ProblemNotice(test_name='Spec XML parsing', location=location,
                                                severity=ProblemSeverity.WARNING, problem='Unable to find access element'))
            return (None, None, None)

        try:
            read_access = str_to_access_type(access_xml.attrib['readPrivilege'])
        except KeyError:
            read_access = ACCESS_CONTROL_PRIVILEGE_ENUM.kUnknownEnumValue
        try:
            write_access = str_to_access_type(access_xml.attrib['writePrivilege'])
        except KeyError:
            write_access = ACCESS_CONTROL_PRIVILEGE_ENUM.kUnknownEnumValue
        try:
            invoke_access = str_to_access_type(access_xml.attrib['invokePrivilege'])
        except KeyError:
            invoke_access = ACCESS_CONTROL_PRIVILEGE_ENUM.kUnknownEnumValue

        return (read_access, write_access, invoke_access)

    def _parse_basic_field_attributes(self, xml_field: ElementTree.Element, component_type: DataTypeEnum, component_tags: dict) -> tuple[str, uint] | None:
        """
        Extract basic field attributes (name and ID) from XML element.

        Args:
            xml_field: XML element representing the field/item/bitfield
            component_type: Type of component being parsed (struct/enum/bitmap)
            component_tags: Mapping of component types to their XML tag info

        Returns:
            Tuple of (name, id) if successful, None if parsing failed
        """
        try:
            name = xml_field.attrib['name']
            # base=0 enables automatic number format detection: "10" -> 10, "0x10" -> 16, etc.
            # This handles XML attributes that can be in decimal, hex, octal, or binary format
            id = uint(int(xml_field.attrib[component_tags[component_type].id_attrib], 0))
            return (name, id)
        except (KeyError, ValueError):
            return None

    @staticmethod
    def _isOptionalField(xml_field: ElementTree.Element) -> bool:
        """
        Determine if a field is optional based on XML attributes and child elements.

        Checks both:
        1. isOptional attribute on the field element
        2. Presence of optionalConform child element

        Args:
            xml_field: XML element representing the field

        Returns:
            True if field is optional, False otherwise
        """
        # Check isOptional attribute
        if 'isOptional' in xml_field.attrib and xml_field.attrib['isOptional'].lower() == 'true':
            return True

        # Check for optionalConform child element
        optional_conform = xml_field.find('./optionalConform')
        return optional_conform is not None

    @staticmethod
    def _isNullableField(xml_field: ElementTree.Element) -> bool:
        """
        Determine if a field is nullable based on XML attributes and child elements.

        Checks both:
        1. isNullable attribute on the field element
        2. nullable attribute on quality child element

        Args:
            xml_field: XML element representing the field

        Returns:
            True if field is nullable, False otherwise
        """
        # Check isNullable attribute
        if 'isNullable' in xml_field.attrib and xml_field.attrib['isNullable'].lower() == 'true':
            return True

        # Check quality child element for nullable attribute
        quality = xml_field.find('./quality')
        return quality is not None and 'nullable' in quality.attrib and quality.attrib['nullable'].lower() == 'true'

    def _parse_field_constraints(self, xml_field: ElementTree.Element) -> dict | None:
        """
        Parse constraint information from XML field element.

        Handles both direct constraint attributes (min/max) and child elements (maxCount).
        For maxCount, also extracts attribute references if present.

        Args:
            xml_field: XML element representing the field

        Returns:
            Dictionary of constraints if any found, None otherwise
        """
        constraint_elements = xml_field.findall('./constraint')
        if not constraint_elements:
            return None

        constraints = {}
        for constraint in constraint_elements:
            # Handle direct attributes like min/max
            for attr_name in ['min', 'max']:
                if attr_name in constraint.attrib:
                    constraints[attr_name] = constraint.attrib[attr_name]

            # Handle maxCount child element
            max_count = constraint.find('./maxCount')
            if max_count is not None and max_count.text is not None:
                constraints['maxCount'] = max_count.text
                # If maxCount references an attribute, store that reference
                attr_element = max_count.find('./attribute')
                if attr_element is not None and 'name' in attr_element.attrib:
                    constraints['maxCountAttribute'] = attr_element.attrib['name']

        # Return None instead of {} to clearly distinguish "no constraints found" from "empty constraints"
        # This matches the Optional[dict] type in XmlDataTypeComponent.constraints
        return constraints if constraints else None

    def _parse_field_conformance(self, xml_field: ElementTree.Element) -> ConformanceCallable:
        """
        Parse conformance information from XML field element with fallback.

        Attempts to parse conformance from XML, but falls back to optional conformance
        if parsing fails. This handles cases where struct fields have arithmetic or
        description conformances that are currently unused.

        Args:
            xml_field: XML element representing the field

        Returns:
            Conformance object (either parsed or optional fallback)
        """
        xml_conformance, problems = get_conformance(xml_field, self._cluster_id)

        # Early return if problems found during conformance extraction
        if problems:
            return optional()

        # Try to parse conformance
        conformance = self.parse_conformance(xml_conformance)
        if not conformance:
            # Fallback to optional conformance
            # Note: Many struct fields have arithmetic/desc conformances that are unused
            return optional()

        return conformance

    def _parse_components(self, struct: ElementTree.Element, component_type: DataTypeEnum) -> dict[uint, XmlDataTypeComponent]:
        """
        Parse components (fields/items/bitfields) from a data type XML element.

        This method orchestrates the parsing of struct fields, enum items, or bitmap bitfields
        by delegating specific parsing tasks to focused helper methods. It handles:
        - Basic attribute extraction (name, ID)
        - Field property detection (optional, nullable)
        - Constraint parsing (min/max values, counts)
        - Conformance parsing with fallback logic

        Args:
            struct: XML element containing the data type definition
            component_type: Type of components to parse (struct/enum/bitmap)

        Returns:
            Dictionary mapping component IDs to XmlDataTypeComponent objects
        """
        @dataclass
        class ComponentTag:
            tag: str
            id_attrib: str
        component_tags = {DataTypeEnum.kStruct: ComponentTag('field', 'id'), DataTypeEnum.kEnum: ComponentTag(
            'item', 'value'), DataTypeEnum.kBitmap: ComponentTag('bitfield', 'bit')}
        components = {}
        struct_name = struct.attrib['name']
        location = ClusterPathLocation(0, int(self._cluster_id) if self._cluster_id is not None else 0)
        for xml_field in list(struct):
            if xml_field.tag != component_tags[component_type].tag:
                continue

            # Parse basic field attributes (name and ID)
            field_attrs = self._parse_basic_field_attributes(xml_field, component_type, component_tags)
            if field_attrs is None:
                p = ProblemNotice("Spec XML Parsing", location=location,
                                  severity=ProblemSeverity.WARNING, problem=f"{component_type.value.capitalize()} field in {struct_name} with no id or name")
                self._problems.append(p)
                continue

            name, id = field_attrs

            # Extract additional field attributes
            summary = xml_field.attrib.get('summary', None)
            type_info = xml_field.attrib.get('type', None) if component_type == DataTypeEnum.kStruct else None

            # Check for duplicate IDs to detect invalid XML data
            if id in components:
                p = ProblemNotice("Spec XML Parsing", location=location,
                                  severity=ProblemSeverity.WARNING,
                                  problem=f"Duplicate {component_type.value} ID {id} in {struct_name} - overwriting previous entry")
                self._problems.append(p)

            # Determine field properties using helper methods
            is_optional = self._isOptionalField(xml_field)
            is_nullable = self._isNullableField(xml_field)
            constraints = self._parse_field_constraints(xml_field)
            conformance = self._parse_field_conformance(xml_field)

            # Create component with all extracted attributes
            components[id] = XmlDataTypeComponent(
                value=id,
                name=name,
                conformance=conformance,
                summary=summary,
                type_info=type_info,
                is_optional=is_optional,
                is_nullable=is_nullable,
                constraints=constraints
            )
        return components

    def _parse_data_type(self, data_type: DataTypeEnum) -> dict[str, XmlDataType]:
        ''' Returns XmlStructs, key is the name.'''
        data_types = {}
        container_tags = self._cluster.iter('dataTypes')
        for container in container_tags:
            xmls = container.iter(str(data_type))
            for element in xmls:
                try:
                    name = element.attrib['name']
                except KeyError:
                    location = ClusterPathLocation(0, int(self._cluster_id) if self._cluster_id is not None else 0)
                    self._problems.append(ProblemNotice("Spec XML Parsing", location=location,
                                          severity=ProblemSeverity.WARNING, problem=f"{data_type.value.capitalize()} {element} with no name"))
                    continue

                # Check for duplicate names to detect invalid XML data
                if name in data_types:
                    location = ClusterPathLocation(0, int(self._cluster_id) if self._cluster_id is not None else 0)
                    self._problems.append(ProblemNotice("Spec XML Parsing", location=location,
                                          severity=ProblemSeverity.WARNING,
                                          problem=f"Duplicate {data_type.value} name '{name}' - overwriting previous entry"))

                # Ensure we're using a valid cluster ID list, never [None]
                cluster_ids = [self._cluster_id] if self._cluster_id is not None else []

                data_types[name] = XmlDataType(
                    data_type=data_type,
                    name=name,
                    components=self._parse_components(element, data_type),
                    cluster_ids=cluster_ids
                )
        return data_types

    def parse_features(self) -> dict[uint, XmlFeature]:
        features = {}
        for element, conformance_xml, _ in self.feature_elements:
            mask = uint(1 << int(element.attrib['bit'], 0))
            conformance = self.parse_conformance(conformance_xml)
            if conformance is None:
                continue
            features[mask] = XmlFeature(code=element.attrib['code'], name=element.attrib['name'],
                                        conformance=conformance)
        return features

    def parse_attributes(self) -> dict[uint, XmlAttribute]:
        attributes: dict[uint, XmlAttribute] = {}
        for element, conformance_xml, access_xml in self.attribute_elements:
            code = uint(int(element.attrib['id'], 0))
            # Some deprecated attributes don't have their types included, for now, lets just fallback to UNKNOWN
            try:
                datatype = element.attrib['type']
            except KeyError:
                datatype = 'UNKNOWN'
            conformance = self.parse_conformance(conformance_xml)
            if conformance is None:
                continue
            if code in attributes:
                # This is one of those fun ones where two different rows have the same id and name, but differ in conformance and ranges
                # I don't have a good way to relate the ranges to the conformance, but they're both acceptable, so let's just or them.
                conformance = or_operation([conformance, attributes[code].conformance])
            read_access, write_access, _ = self.parse_access(element, access_xml, conformance)
            write_optional = False
            if write_access not in [None, ACCESS_CONTROL_PRIVILEGE_ENUM.kUnknownEnumValue]:
                write_optional = self.parse_write_optional(element, access_xml)
            attributes[code] = XmlAttribute(name=element.attrib['name'], datatype=datatype,
                                            conformance=conformance,
                                            read_access=get_access_privilege_or_unknown(read_access),
                                            write_access=get_access_privilege_or_unknown(write_access),
                                            write_optional=write_optional)
        # Add in the global attributes for the base class
        for id in GlobalAttributeIds:
            # TODO: Add data type here. Right now it's unused. We should parse this from the spec.
            attributes[uint(id)] = XmlAttribute(name=id.to_name(), datatype="", conformance=mandatory(
            ), read_access=ACCESS_CONTROL_PRIVILEGE_ENUM.kView, write_access=ACCESS_CONTROL_PRIVILEGE_ENUM.kUnknownEnumValue, write_optional=False)
        return attributes

    def get_command_type(self, element: ElementTree.Element) -> CommandType:
        try:
            if element.attrib['direction'].lower() == 'responsefromserver':
                return CommandType.GENERATED
            if element.attrib['direction'].lower() == 'commandtoclient':
                return CommandType.UNKNOWN
            if element.attrib['direction'].lower() == 'commandtoserver':
                return CommandType.ACCEPTED
            if element.attrib['direction'].lower() == 'responsefromclient':
                return CommandType.UNKNOWN
            raise Exception(f"Unknown direction: {element.attrib['direction']}")
        except KeyError:
            return CommandType.UNKNOWN

    def parse_unknown_commands(self) -> list[XmlCommand]:
        commands = []
        for element, conformance_xml, access_xml in self.command_elements:
            if self.get_command_type(element) != CommandType.UNKNOWN:
                continue
            code = int(element.attrib['id'], 0)
            conformance = self.parse_conformance(conformance_xml)

            if conformance is not None:
                _, _, privilege = self.parse_access(element, access_xml, conformance)
                commands.append(XmlCommand(id=code, name=element.attrib['name'], conformance=conformance,
                                           privilege=get_access_privilege_or_unknown(privilege)))
        return commands

    def parse_commands(self, command_type: CommandType) -> dict[uint, XmlCommand]:
        commands: dict[uint, XmlCommand] = {}
        for element, conformance_xml, access_xml in self.command_elements:
            if self.get_command_type(element) != command_type:
                continue
            code = uint(int(element.attrib['id'], 0))
            conformance = self.parse_conformance(conformance_xml)
            if conformance is None:
                continue
            if code in commands:
                conformance = or_operation([conformance, commands[code].conformance])

            _, _, privilege = self.parse_access(element, access_xml, conformance)
            commands[uint(code)] = XmlCommand(id=code, name=element.attrib['name'], conformance=conformance,
                                              privilege=get_access_privilege_or_unknown(privilege))
        return commands

    def parse_events(self) -> dict[uint, XmlEvent]:
        events: dict[uint, XmlEvent] = {}
        for element, conformance_xml, access_xml in self.event_elements:
            code = uint(int(element.attrib['id'], 0))
            conformance = self.parse_conformance(conformance_xml)
            if conformance is None:
                continue
            if code in events:
                conformance = or_operation([conformance, events[code].conformance])
            events[code] = XmlEvent(name=element.attrib['name'], conformance=conformance)
        return events

    def create_cluster(self) -> XmlCluster:
        try:
            revision = int(self._cluster.attrib['revision'], 0)
        except ValueError:
            revision = 0
        return XmlCluster(revision=revision, derived=self._derived,
                          name=self._name, feature_map=self.params.feature_map,
                          attribute_map=self.params.attribute_map, command_map=self.params.command_map,
                          features=self.parse_features(),
                          attributes=self.parse_attributes(),
                          accepted_commands=self.parse_commands(CommandType.ACCEPTED),
                          generated_commands=self.parse_commands(CommandType.GENERATED),
                          unknown_commands=self.parse_unknown_commands(),
                          events=self.parse_events(),
                          structs=self._parse_data_type(DataTypeEnum.kStruct),
                          enums=self._parse_data_type(DataTypeEnum.kEnum),
                          bitmaps=self._parse_data_type(DataTypeEnum.kBitmap),
                          pics=self._pics if self._pics is not None else "", is_provisional=self._is_provisional)

    def get_problems(self) -> list[ProblemNotice]:
        return self._problems


def add_cluster_data_from_xml(xml: ElementTree.Element, clusters: dict[uint, XmlCluster], pure_base_clusters: dict[str, XmlCluster], ids_by_name: dict[str, uint], problems: list[ProblemNotice]) -> None:
    ''' Adds cluster data to the supplied dicts as appropriate

        xml: XML element read from from the XML cluster file
        clusters: dict of id -> XmlCluster. This function will append new clusters as appropriate to this dict.
        pure_base_clusters: dict of base name -> XmlCluster. This data structure is used to hold pure base clusters that don't have
                            an ID. This function will append new pure base clusters as appropriate to this dict.
        ids_by_name: dict of cluster name -> ID. This function will append new IDs as appropriate to this dict.
        problems: list of any problems encountered during spec parsing. This function will append problems as appropriate to this list.
    '''
    cluster = xml.iter('cluster')
    for c in cluster:
        ids = c.iter('clusterId')
        for id in ids:
            name = id.get('name')
            cluster_id_str = id.get('id')
            cluster_id: Optional[uint] = None
            if cluster_id_str:
                cluster_id = uint(int(cluster_id_str, 0))

            if name is None:
                location = ClusterPathLocation(endpoint_id=0, cluster_id=0 if cluster_id is None else cluster_id)
                problems.append(ProblemNotice(test_name="Spec XML parsing", location=location,
                                severity=ProblemSeverity.WARNING, problem=f"Cluster with no name {c}"))
                continue

            parser = ClusterParser(c, cluster_id, name)
            new = parser.create_cluster()
            problems = problems + parser.get_problems()

            if cluster_id:
                clusters[cluster_id] = new
                ids_by_name[name] = cluster_id
            else:
                # Fully derived clusters have no id, but also shouldn't appear on a device.
                # We do need to keep them, though, because we need to update the derived
                # clusters. We keep them in a special dict by name, so they can be thrown
                # away later.
                pure_base_clusters[name] = new


def check_clusters_for_unknown_commands(clusters: dict[uint, XmlCluster], problems: list[ProblemNotice]):
    for id, cluster in clusters.items():
        for cmd in cluster.unknown_commands:
            problems.append(ProblemNotice(test_name="Spec XML parsing", location=CommandPathLocation(
                endpoint_id=0, cluster_id=id, command_id=cmd.id), severity=ProblemSeverity.WARNING, problem="Command with unknown direction"))


class PrebuiltDataModelDirectory(Enum):
    k1_2 = auto()
    k1_3 = auto()
    k1_4 = auto()
    k1_4_1 = auto()
    k1_4_2 = auto()
    k1_5 = auto()

    @property
    def dirname(self):
        if self == PrebuiltDataModelDirectory.k1_2:
            return "1.2"
        if self == PrebuiltDataModelDirectory.k1_3:
            return "1.3"
        if self == PrebuiltDataModelDirectory.k1_4:
            return "1.4"
        if self == PrebuiltDataModelDirectory.k1_4_1:
            return "1.4.1"
        if self == PrebuiltDataModelDirectory.k1_4_2:
            return "1.4.2"
        if self == PrebuiltDataModelDirectory.k1_5:
            return "1.5"
        raise KeyError("Invalid enum: %r" % self)


class DataModelLevel(Enum):
    kCluster = auto()
    kDeviceType = auto()
    kGlobal = auto()
    kNamespace = auto()

    @property
    def dirname(self):
        if self == DataModelLevel.kCluster:
            return "clusters"
        if self == DataModelLevel.kDeviceType:
            return "device_types"
        if self == DataModelLevel.kGlobal:
            return "globals"
        if self == DataModelLevel.kNamespace:
            return "namespaces"
        raise KeyError("Invalid enum: %r" % self)


def get_data_model_directory(data_model_directory: Union[PrebuiltDataModelDirectory, Traversable], data_model_level: DataModelLevel = DataModelLevel.kCluster) -> Traversable:
    """
    Get the directory of the data model for a specific version and level from the installed package.

    `data_model_directory` given as a path MUST be of type Traversable (often `pathlib.Path(somepathstring)`).
    If `data_model_directory` is given as a Traversable, it is returned directly WITHOUT using the data_model_level at all.
    """
    # Early return if data_model_directory is already a Traversable type
    if not isinstance(data_model_directory, PrebuiltDataModelDirectory):
        # data_model_directory is a Traversable (e.g. pathlib.Path to an extracted root)
        # Return directly as per the docstring - it should already contain the correct directory structure
        return data_model_directory

    # If it's a prebuilt directory, build the path based on the version and data model level
    zip_file_traversable = pkg_resources.files(importlib.import_module('matter.testing')).joinpath(
        'data_model').joinpath(data_model_directory.dirname).joinpath('allfiles.zip')

    # Avoid returning a zipfile.Path backed by a closed file handle. Build Path from the filesystem path
    # so the ZipFile lifecycle is managed by zipfile.Path itself.
    # mypy: Traversable does not declare __fspath__, but runtime object from importlib.resources
    # is a FileSystem resource that implements it. Safe to coerce for zipfile.Path usage.
    zip_path = os.fspath(zip_file_traversable)  # type: ignore[call-overload]
    zip_root = zipfile.Path(zip_path)
    return zip_root / data_model_level.dirname


def build_xml_clusters(data_model_directory: Union[PrebuiltDataModelDirectory, Traversable]) -> typing.Tuple[dict[uint, XmlCluster], list[ProblemNotice]]:
    """
    Build XML clusters from the specified data model directory.
    This function supports both pre-built locations and full paths.

    `data_model_directory`` given as a path MUST be of type Traversable (often `pathlib.Path(somepathstring)`).
    If data_model_directory is a Traversable, it is assumed to already contain `clusters` (i.e. be a directory
    with all XML files in it)
    """

    clusters: dict[uint, XmlCluster] = {}
    pure_base_clusters: dict[str, XmlCluster] = {}
    ids_by_name: dict[str, uint] = {}
    problems: list[ProblemNotice] = []

    top = get_data_model_directory(data_model_directory, DataModelLevel.kCluster)
    LOGGER.info("Reading XML clusters from %r", top)

    found_xmls = 0
    for f in top.iterdir():
        if not f.name.endswith('.xml'):
            LOGGER.info("Ignoring non-XML file %s", f.name)
            continue

        found_xmls += 1
        with f.open("r", encoding="utf8") as file:
            root = ElementTree.parse(file).getroot()
            add_cluster_data_from_xml(root, clusters, pure_base_clusters, ids_by_name, problems)

    # For now we assume even a single XML means the directory was probably OK
    # we may increase this later as most our data model directories are larger
    #
    # Intent here is to make user aware of typos in paths instead of silently having
    # empty parsing
    if found_xmls < 1:
        raise SpecParsingException(f'No data model files found in specified directory {top:!r}')

    # There are a few clusters where the conformance columns are listed as desc. These clusters need specific, targeted tests
    # to properly assess conformance. Here, we list them as Optional to allow these for the general test. Targeted tests are described below.
    # Descriptor - TagList feature - this feature is mandated when the duplicate condition holds for the endpoint. It is tested in DESC-2.2
    # Actions cluster - all commands - these need to be listed in the ActionsList attribute to be supported.
    #                                  We do not currently have a test for this. Please see https://github.com/CHIP-Specifications/chip-test-plans/issues/3646.

    def remove_problem(location: typing.Union[CommandPathLocation, FeaturePathLocation]):
        nonlocal problems
        problems = [p for p in problems if p.location != location]

    descriptor_id = uint(Clusters.Descriptor.id)
    code = 'TAGLIST'
    mask = clusters[descriptor_id].feature_map[code]
    clusters[descriptor_id].features[mask].conformance = optional()
    remove_problem(FeaturePathLocation(endpoint_id=0, cluster_id=descriptor_id, feature_code=code))

    action_id = uint(Clusters.Actions.id)
    for c in Clusters.ClusterObjects.ALL_ACCEPTED_COMMANDS[action_id]:
        clusters[action_id].accepted_commands[c].conformance = optional()
        remove_problem(CommandPathLocation(endpoint_id=0, cluster_id=action_id, command_id=c))

    combine_derived_clusters_with_base(clusters, pure_base_clusters, ids_by_name, problems)

    # TODO: All these fixups should be removed BEFORE SVE if at all possible
    # Workaround for Color Control cluster - the spec uses a non-standard conformance. Set all to optional now, will need
    # to implement either arithmetic conformance handling (once spec changes land here) or specific test
    # https://github.com/CHIP-Specifications/connectedhomeip-spec/pull/7808 for spec changes.
    # see 3.2.8. Defined Primaries Information Attribute Set, affects Primary<#>X/Y/Intensity attributes.
    cc_id = uint(Clusters.ColorControl.id)
    cc_attr = Clusters.ColorControl.Attributes
    affected_attributes = [cc_attr.Primary1X,
                           cc_attr.Primary1Y,
                           cc_attr.Primary1Intensity,
                           cc_attr.Primary2X,
                           cc_attr.Primary2Y,
                           cc_attr.Primary2Intensity,
                           cc_attr.Primary3X,
                           cc_attr.Primary3Y,
                           cc_attr.Primary3Intensity,
                           cc_attr.Primary4X,
                           cc_attr.Primary4Y,
                           cc_attr.Primary4Intensity,
                           cc_attr.Primary5X,
                           cc_attr.Primary5Y,
                           cc_attr.Primary5Intensity,
                           cc_attr.Primary6X,
                           cc_attr.Primary6Y,
                           cc_attr.Primary6Intensity,
                           ]
    for a in affected_attributes:
        clusters[cc_id].attributes[a.attribute_id].conformance = optional()

    # Workaround for temp control cluster - this is parsed incorrectly in the DM XML and is missing all its attributes
    # Remove this workaround when https://github.com/csa-data-model/projects/issues/330 is fixed
    temp_control_id = uint(Clusters.TemperatureControl.id)
    if temp_control_id in clusters and not clusters[temp_control_id].attributes:
        view = ACCESS_CONTROL_PRIVILEGE_ENUM.kView
        none = ACCESS_CONTROL_PRIVILEGE_ENUM.kUnknownEnumValue
        clusters[temp_control_id].attributes = {
            uint(0x00): XmlAttribute(name='TemperatureSetpoint', datatype='temperature', conformance=feature(uint(0x01), 'TN'), read_access=view, write_access=none, write_optional=False),
            uint(0x01): XmlAttribute(name='MinTemperature', datatype='temperature', conformance=feature(uint(0x01), 'TN'), read_access=view, write_access=none, write_optional=False),
            uint(0x02): XmlAttribute(name='MaxTemperature', datatype='temperature', conformance=feature(uint(0x01), 'TN'), read_access=view, write_access=none, write_optional=False),
            uint(0x03): XmlAttribute(name='Step', datatype='temperature', conformance=feature(uint(0x04), 'STEP'), read_access=view, write_access=none, write_optional=False),
            uint(0x04): XmlAttribute(name='SelectedTemperatureLevel', datatype='uint8', conformance=feature(uint(0x02), 'TL'), read_access=view, write_access=none, write_optional=False),
            uint(0x05): XmlAttribute(name='SupportedTemperatureLevels', datatype='list', conformance=feature(uint(0x02), 'TL'), read_access=view, write_access=none, write_optional=False),
        }

    # TODO: Need automated parsing for atomic attributes.
    atomic_request_cmd_id = uint(0xFE)
    atomic_response_cmd_id = uint(0xFD)
    atomic_request_name = "Atomic Request"
    atomic_response_name = "Atomic Response"
    presets_name = "Presets"
    schedules_name = "Schedules"
    thermostat_id = uint(Clusters.Thermostat.id)
    if clusters[thermostat_id].revision >= 8:
        presents_id = uint(clusters[thermostat_id].attribute_map[presets_name])
        schedules_id = uint(clusters[thermostat_id].attribute_map[schedules_name])
        conformance = or_operation([conformance_support.attribute(presents_id, presets_name),
                                    conformance_support.attribute(schedules_id, schedules_name)])

        clusters[thermostat_id].accepted_commands[atomic_request_cmd_id] = XmlCommand(
            id=atomic_request_cmd_id, name=atomic_request_name, conformance=conformance, privilege=ACCESS_CONTROL_PRIVILEGE_ENUM.kOperate)
        clusters[thermostat_id].generated_commands[atomic_response_cmd_id] = XmlCommand(
            id=atomic_response_cmd_id, name=atomic_response_name, conformance=conformance, privilege=ACCESS_CONTROL_PRIVILEGE_ENUM.kOperate)
        clusters[thermostat_id].command_map[atomic_request_name] = atomic_request_cmd_id
        clusters[thermostat_id].command_map[atomic_response_name] = atomic_response_cmd_id

    check_clusters_for_unknown_commands(clusters, problems)

    return clusters, problems


def combine_derived_clusters_with_base(xml_clusters: dict[uint, XmlCluster], pure_base_clusters: dict[str, XmlCluster], ids_by_name: dict[str, uint], problems: list[ProblemNotice]) -> None:
    ''' Overrides base elements with the derived cluster values for derived clusters. '''

    def combine_attributes(base: dict[uint, XmlAttribute], derived: dict[uint, XmlAttribute], cluster_id: uint, problems: list[ProblemNotice]) -> dict[uint, XmlAttribute]:
        ret = deepcopy(base)
        extras = {k: v for k, v in derived.items() if k not in base}
        overrides = {k: v for k, v in derived.items() if k in base}
        ret.update(extras)
        for id, override in overrides.items():
            if override.conformance is not None:
                ret[id].conformance = override.conformance
            if override.read_access:
                ret[id].read_access = override.read_access
            if override.write_access:
                ret[id].write_access = override.write_access

        for attr_id, attribute in ret.items():
            if attribute.read_access == ACCESS_CONTROL_PRIVILEGE_ENUM.kUnknownEnumValue and \
               attribute.write_access == ACCESS_CONTROL_PRIVILEGE_ENUM.kUnknownEnumValue:
                location = AttributePathLocation(endpoint_id=0, cluster_id=cluster_id, attribute_id=attr_id)
                problems.append(ProblemNotice(test_name='Spec XML parsing', location=location,
                                              severity=ProblemSeverity.WARNING, problem=f'Attribute {attribute.name} (ID: {attr_id}) in cluster {cluster_id} has unknown read and write access after combining base and derived values.'))
        return ret

    # We have the information now about which clusters are derived, so we need to fix them up. Apply first the base cluster,
    # then add the specific cluster overtop
    for id, c in xml_clusters.items():
        if c.derived:
            base_name = c.derived
            if base_name in ids_by_name:
                base = xml_clusters[ids_by_name[c.derived]]
            else:
                base = pure_base_clusters[base_name]

            feature_map = deepcopy(base.feature_map)
            feature_map.update(c.feature_map)
            attribute_map = deepcopy(base.attribute_map)
            attribute_map.update(c.attribute_map)
            command_map = deepcopy(base.command_map)
            command_map.update(c.command_map)
            features = deepcopy(base.features)
            features.update(c.features)
            attributes = combine_attributes(base.attributes, c.attributes, id, problems)
            accepted_commands = deepcopy(base.accepted_commands)
            accepted_commands.update(c.accepted_commands)
            generated_commands = deepcopy(base.generated_commands)
            generated_commands.update(c.generated_commands)
            events = deepcopy(base.events)
            events.update(c.events)
            structs = deepcopy(base.structs)
            structs.update(c.structs)
            enums = deepcopy(base.enums)
            enums.update(c.enums)
            bitmaps = deepcopy(base.bitmaps)
            bitmaps.update(c.bitmaps)
            unknown_commands = deepcopy(base.unknown_commands)
            for cmd in c.unknown_commands:
                if cmd.id in accepted_commands and cmd.name == accepted_commands[uint(cmd.id)].name:
                    accepted_commands[uint(cmd.id)].conformance = cmd.conformance
                elif cmd.id in generated_commands and cmd.name == generated_commands[uint(cmd.id)].name:
                    generated_commands[uint(cmd.id)].conformance = cmd.conformance
                else:
                    unknown_commands.append(cmd)
            provisional = c.is_provisional or base.is_provisional

            new = XmlCluster(revision=c.revision, derived=c.derived, name=c.name,
                             feature_map=feature_map, attribute_map=attribute_map, command_map=command_map,
                             features=features, attributes=attributes, accepted_commands=accepted_commands,
                             generated_commands=generated_commands, unknown_commands=unknown_commands, events=events, structs=structs,
                             enums=enums, bitmaps=bitmaps, pics=c.pics, is_provisional=provisional)
            xml_clusters[id] = new


def parse_namespace(et: ElementTree.Element) -> tuple[XmlNamespace, list[ProblemNotice]]:
    """Parse a single namespace XML definition"""
    problems: list[ProblemNotice] = []
    namespace = XmlNamespace()

    # Parse namespace attributes
    namespace_id = et.get('id')
    if namespace_id is not None:
        try:
            namespace.id = int(str(namespace_id), 16)
        except (ValueError, TypeError):
            problems.append(ProblemNotice(
                test_name="Parse Namespace XML",
                location=NamespacePathLocation(),
                severity=ProblemSeverity.WARNING,
                problem=f"Invalid namespace ID: {namespace_id}"
            ))
    else:
        problems.append(ProblemNotice(
            test_name="Parse Namespace XML",
            location=NamespacePathLocation(),
            severity=ProblemSeverity.WARNING,
            problem="Missing namespace ID"
        ))

    # Parse and validate namespace name
    namespace.name = et.get('name', '').strip()
    if not namespace.name:
        problems.append(ProblemNotice(
            test_name="Parse Namespace XML",
            location=NamespacePathLocation(namespace_id=getattr(namespace, 'id', None)),
            severity=ProblemSeverity.WARNING,
            problem="Missing or empty namespace name"
        ))

    # Parse tags
    tags_elem = et.find('tags')
    if tags_elem is not None:
        for tag_elem in tags_elem.findall('tag'):
            tag = XmlTag()
            tag_id = tag_elem.get('id')
            if tag_id is not None:
                try:
                    tag.id = int(str(tag_id), 0)
                except (ValueError, TypeError):
                    problems.append(ProblemNotice(
                        test_name="Parse Namespace XML",
                        location=NamespacePathLocation(namespace_id=namespace.id),
                        severity=ProblemSeverity.WARNING,
                        problem=f"Invalid tag ID: {tag_id}"
                    ))
                    continue

            tag.name = tag_elem.get('name', '').strip()
            if not tag.name:
                problems.append(ProblemNotice(
                    test_name="Parse Namespace XML",
                    location=NamespacePathLocation(namespace_id=namespace.id, tag_id=getattr(tag, 'id', None)),
                    severity=ProblemSeverity.WARNING,
                    problem=f"Missing name for tag {tag.id}"
                ))
                continue

            desc_elem = tag_elem.find('description')
            if desc_elem is not None and desc_elem.text:
                tag.description = desc_elem.text.strip()

            namespace.tags[tag.id] = tag

    return namespace, problems


def build_xml_namespaces(data_model_directory: typing.Union[PrebuiltDataModelDirectory, Traversable]) -> tuple[dict[int, XmlNamespace], list[ProblemNotice]]:
    """Build a dictionary of namespaces from XML files in the given directory"""
    namespace_dir = get_data_model_directory(data_model_directory, DataModelLevel.kNamespace)
    namespaces: dict[int, XmlNamespace] = {}
    problems: list[ProblemNotice] = []

    found_xmls = 0

    try:
        for filename in namespace_dir.iterdir():
            if not filename.name.endswith('.xml'):
                continue
            LOGGER.info('Parsing file %s', str(filename))
            found_xmls += 1

            try:
                with filename.open('r', encoding="utf8") as xml:
                    root = ElementTree.parse(xml).getroot()
                    namespace, parse_problems = parse_namespace(root)
                    problems.extend(parse_problems)

                    if namespace.id in namespaces:
                        problems.append(ProblemNotice(
                            test_name="Build XML Namespaces",
                            location=NamespacePathLocation(namespace_id=namespace.id),
                            severity=ProblemSeverity.WARNING,
                            problem=f"Duplicate namespace ID 0x{namespace.id:04X} in {filename.name}"
                        ))
                    else:
                        namespaces[namespace.id] = namespace

            except Exception as e:
                problems.append(ProblemNotice(
                    test_name="Build XML Namespaces",
                    location=UnknownProblemLocation(),
                    severity=ProblemSeverity.WARNING,
                    problem=f"Failed to parse {filename.name}: {str(e)}"
                ))

    except Exception as e:
        problems.append(ProblemNotice(
            test_name="Build XML Namespaces",
            location=UnknownProblemLocation(),
            severity=ProblemSeverity.WARNING,
            problem=f"Failed to access namespace directory: {str(e)}"
        ))

    if found_xmls < 1:
        LOGGER.warning("No XML files found in the specified namespace directory: %r", namespace_dir)
        problems.append(ProblemNotice(
            test_name="Build XML Namespaces",
            location=UnknownProblemLocation(),
            severity=ProblemSeverity.WARNING,
            problem=f"No XML files found in namespace directory: {str(namespace_dir)}"
        ))

    # Print problems for debugging
    if problems:
        LOGGER.warning("Found %d problems while parsing namespaces:", len(problems))
        for problem in problems:
            LOGGER.warning("  - %s", str(problem))

    return namespaces, problems


def parse_single_device_type(root: ElementTree.Element, cluster_definition_xml: dict[uint, XmlCluster]) -> tuple[dict[int, XmlDeviceType], list[ProblemNotice]]:
    problems: list[ProblemNotice] = []
    device_types: dict[int, XmlDeviceType] = {}
    d = root
    if d.tag == 'deviceType':
        device_name = d.attrib['name']
        location = DeviceTypePathLocation(device_type_id=0)

        str_id = d.attrib.get('id', "")
        if not str_id:
            if device_name == "Base Device Type":
                # Base is special device type, we're going to call it -1 so we can combine and remove it later.
                str_id = '-1'
            else:
                problems.append(ProblemNotice("Parse Device Type XML", location=location,
                                severity=ProblemSeverity.WARNING, problem=f"Device type {device_name} does not have an ID listed"))
                return device_types, problems
        try:
            id = int(str_id, 0)
            revision = int(d.attrib['revision'], 0)
        except ValueError:
            problems.append(ProblemNotice("Parse Device Type XML", location=location,
                            severity=ProblemSeverity.WARNING,
                            problem=f"Device type {device_name} does not a valid ID or revision. ID: {str_id} revision: {d.get('revision', 'UNKNOWN')}"))
            return device_types, problems
        if id in DEVICE_TYPE_NAME_FIXES:
            device_name = DEVICE_TYPE_NAME_FIXES[id]

        location = DeviceTypePathLocation(device_type_id=id)

        try:
            classification = next(d.iter('classification'))
            scope = classification.attrib['scope']
            device_class = classification.attrib['class']
            superset_of_device_type_name = classification.attrib.get('superset', None)
        except (KeyError, StopIteration):
            # this is fine for base device type
            if id == -1:
                scope = 'BASE'
                device_class = 'BASE'
                superset_of_device_type_name = None
            else:
                location = DeviceTypePathLocation(device_type_id=id)
                problems.append(ProblemNotice("Parse Device Type XML", location=location,
                                severity=ProblemSeverity.WARNING, problem="Unable to find classification data for device type"))
                return device_types, problems
        device_types[id] = XmlDeviceType(name=device_name, revision=revision, server_clusters={}, client_clusters={},
                                         classification_class=device_class, classification_scope=scope, superset_of_device_type_name=superset_of_device_type_name)
        try:
            main_endpoint_clusters = next(d.iter('clusters'))
            clusters = main_endpoint_clusters.findall('cluster')
        except StopIteration:
            # no clusters in this device type - children device types only
            clusters = []
        for c in clusters:
            try:
                try:
                    cid = uint(int(c.attrib['id'], 0))
                except ValueError:
                    location = DeviceTypePathLocation(device_type_id=id)
                    problems.append(ProblemNotice("Parse Device Type XML", location=location,
                                    severity=ProblemSeverity.WARNING, problem=f"Unknown cluster id {c.attrib['id']}"))
                    continue
                # Workaround for 1.3 device types with zigbee clusters and old scenes
                # This is OK because there are other tests that ensure that unknown clusters do not appear on the device
                if cid not in cluster_definition_xml:
                    LOGGER.info(f"Skipping unknown cluster {cid:04X}")
                    continue
                conformance_xml, tmp_problem = get_conformance(c, cid)
                if tmp_problem:
                    problems.append(tmp_problem)
                    continue
                cluster_conformance_params = ConformanceParseParameters(
                    feature_map=cluster_definition_xml[cid].feature_map, attribute_map=cluster_definition_xml[cid].attribute_map, command_map=cluster_definition_xml[cid].command_map)
                conformance = parse_callable_from_xml(conformance_xml, cluster_conformance_params)
                side_dict = {'server': ClusterSide.SERVER, 'client': ClusterSide.CLIENT}
                side = side_dict[c.attrib['side']]
                cluster_name = c.attrib['name']
                if cid in CLUSTER_NAME_FIXES:
                    cluster_name = CLUSTER_NAME_FIXES[cid]
                cluster = XmlDeviceTypeClusterRequirements(name=cluster_name, side=side, conformance=conformance)

                def append_overrides(override_element_type: str):
                    if override_element_type == 'feature':
                        # The device types use feature name rather than feature code. So we need to build a new map.
                        name_to_id_map = {f.name: id for id, f in cluster_definition_xml[cid].features.items()}
                        # But also...that's not universal, so let's be tolerant to using the code too.
                        name_to_id_map.update(cluster_definition_xml[cid].feature_map)
                        override = cluster.feature_overrides
                    elif override_element_type == 'attribute':
                        name_to_id_map = cluster_definition_xml[cid].attribute_map
                        override = cluster.attribute_overrides
                    elif override_element_type == 'command':
                        name_to_id_map = cluster_definition_xml[cid].command_map
                        override = cluster.command_overrides
                    else:
                        problems.append(ProblemNotice("Parse Device Type XML", location=location,
                                        severity=ProblemSeverity.WARNING, problem=f"Request for unknown element override type {override_element_type} - this is a script error"))
                        return

                    container = c.find(f'{override_element_type}s')
                    if container is None:
                        return

                    elements = container.iter(override_element_type)
                    for e in elements:
                        try:
                            element_name = e.attrib['name']
                        except KeyError:
                            if override_element_type == 'feature':
                                try:
                                    element_name = e.attrib['code']
                                except KeyError:
                                    element_name = None
                            else:
                                element_name = None
                        if element_name is None:
                            problems.append(ProblemNotice("Parse Device Type XML", location=location,
                                            severity=ProblemSeverity.WARNING, problem=f"Missing {override_element_type} name for override in cluster 0x{cid:04X}, e={str(e.attrib)}"))
                            continue

                        try:
                            conformance_xml, tmp_problem = get_conformance(e, cid)
                            if tmp_problem:
                                # It's not actually a problem if there is no conformance override - it might be a constraint override. Just continue
                                continue
                            conformance_override = parse_callable_from_xml(conformance_xml, cluster_conformance_params)

                            map_id = [name_to_id_map[n] for n in name_to_id_map if _fuzzy_name(n) ==
                                      _fuzzy_name(element_name)]
                            if len(map_id) == 0:
                                # The thermostat in particular explicitly disallows some zigbee things that don't appear in the spec due to
                                # ifdefs. We can ignore problems if the device type spec disallows things that don't exist.
                                if is_disallowed(conformance_override):
                                    LOGGER.info(
                                        f"Ignoring unknown {override_element_type} {element_name} in cluster {cid} because the conformance is disallowed")
                                    continue
                                problems.append(ProblemNotice("Parse Device Type XML", location=location,
                                                severity=ProblemSeverity.WARNING, problem=f"Unknown {override_element_type} {element_name} in cluster 0x{cid:04X} - map = {map_id}"))
                            else:
                                override[map_id[0]] = conformance_override

                        except ConformanceException as ex:
                            problems.append(ProblemNotice("Parse Device Type XML", location=location,
                                            severity=ProblemSeverity.WARNING, problem=f"Unable to parse {override_element_type} conformance for {element_name} in cluster 0x{cid:04X} - {ex}"))

                append_overrides('feature')
                append_overrides('attribute')
                append_overrides('command')

                if side == ClusterSide.SERVER:
                    device_types[id].server_clusters[cid] = cluster
                else:
                    device_types[id].client_clusters[cid] = cluster

            except ConformanceException as ex:
                location = DeviceTypePathLocation(device_type_id=id, cluster_id=cid)
                problems.append(ProblemNotice("Parse Device Type XML", location=location,
                                severity=ProblemSeverity.WARNING, problem=f"Unable to parse conformance for cluster - {ex}"))
            # NOTE: Spec currently does a bad job of matching these exactly to the names and codes
            # so this will need a bit of fancy handling here to get this right.
    return device_types, problems


def build_xml_device_types(data_model_directory: typing.Union[PrebuiltDataModelDirectory, Traversable], cluster_definition_xml: Optional[dict[uint, XmlCluster]] = None) -> tuple[dict[int, XmlDeviceType], list[ProblemNotice]]:
    top = get_data_model_directory(data_model_directory, DataModelLevel.kDeviceType)
    device_types: dict[int, XmlDeviceType] = {}
    problems: list[ProblemNotice] = []
    if not cluster_definition_xml:
        cluster_dir = data_model_directory
        if not isinstance(data_model_directory, PrebuiltDataModelDirectory):
            # Transform this into the cluster directory
            cluster_dir = data_model_directory.joinpath('..', 'clusters')
        cluster_definition_xml, _ = build_xml_clusters(cluster_dir)

    found_xmls = 0

    for file in top.iterdir():
        if not file.name.endswith('.xml'):
            continue
        found_xmls += 1
        with file.open('r', encoding="utf8") as xml:
            root = ElementTree.parse(xml).getroot()
            tmp_device_types, tmp_problems = parse_single_device_type(root, cluster_definition_xml)
            problems = problems + tmp_problems
            device_types.update(tmp_device_types)

    if found_xmls < 1:
        LOGGER.warning("No XML files found in the specified device type directory: %r", top)

    if -1 not in device_types:
        raise ConformanceException("Base device type not found in device type xml data")

    # Add in the base device type information and remove the base device type from the device_types
    for d in device_types.values():
        d.server_clusters.update(device_types[-1].server_clusters)
    device_types.pop(-1)

    # Fix up supersets
    for id, d in device_types.items():
        def standardize_name(name: str):
            return name.replace(' ', '').replace('/', '').lower()
        if d.superset_of_device_type_name is None:
            continue
        name = standardize_name(d.superset_of_device_type_name)
        matches = [id for id, d in device_types.items() if standardize_name(d.name) == name]
        if len(matches) != 1:
            problems.append(ProblemNotice('Device types parsing', location=DeviceTypePathLocation(
                id), severity=ProblemSeverity.ERROR, problem=f"No unique match found for superset {d.superset_of_device_type_name}"))
            break
        d.superset_of_device_type_id = matches[0]

    return device_types, problems


def build_xml_global_data_types(data_model_directory: Union[PrebuiltDataModelDirectory, Traversable]) -> typing.Tuple[dict[str, dict[str, XmlDataType]], list[ProblemNotice]]:
    """
    Build XML global data types from the globals data model directory.

    `data_model_directory` given as a path MUST be of type Traversable (often `pathlib.Path(somepathstring)`).
    If data_model_directory is a Traversable, it is assumed to already contain `globals`

    Returns:
        Tuple of (global_data_types, problems) where global_data_types is a dict with keys:
        - 'structs': dict[str, XmlDataType] - global structs by name
        - 'enums': dict[str, XmlDataType] - global enums by name
        - 'bitmaps': dict[str, XmlDataType] - global bitmaps by name
    """

    global_data_types: dict[str, dict[str, XmlDataType]] = {
        'structs': {},
        'enums': {},
        'bitmaps': {}
    }
    problems: list[ProblemNotice] = []

    top = get_data_model_directory(data_model_directory, DataModelLevel.kGlobal)
    LOGGER.info("Reading XML global data types from %r", top)

    # Map of XML file names to data type categories
    file_to_datatype_map = {
        'Structs.xml': ('structs', DataTypeEnum.kStruct),
        'Enums.xml': ('enums', DataTypeEnum.kEnum),
        'Bitmaps.xml': ('bitmaps', DataTypeEnum.kBitmap)
    }

    found_xmls = 0
    for f in top.iterdir():
        if not f.name.endswith('.xml'):
            LOGGER.info("Ignoring non-XML file %s", f.name)
            continue

        if f.name not in file_to_datatype_map:
            LOGGER.info("Ignoring XML file %s (not a data type file - only parsing Structs.xml, Enums.xml, Bitmaps.xml)", f.name)
            continue

        found_xmls += 1
        category, data_type_enum = file_to_datatype_map[f.name]

        with f.open("r", encoding="utf8") as file:
            root = ElementTree.parse(file).getroot()

            # Parse each data type element in the file
            for element in root:
                if element.tag != str(data_type_enum):
                    continue

                try:
                    name = element.attrib['name']
                except KeyError:
                    location = ClusterPathLocation(0, 0)  # Global types have no cluster ID
                    problems.append(ProblemNotice("Global Data Type XML Parsing", location=location,
                                                  severity=ProblemSeverity.WARNING,
                                                  problem=f"Global {data_type_enum} with no name in {f.name}"))
                    continue

                # Create a temporary parser to parse components
                # Global data types have no cluster ID, so we pass None
                temp_parser = ClusterParser(ElementTree.Element('cluster'), None, 'GlobalDataTypes')
                components = temp_parser._parse_components(element, data_type_enum)
                temp_problems = temp_parser.get_problems()

                # Filter out conformance parsing problems for global data types
                # Global data types often have complex field reference conformances that our parser can't handle yet
                filtered_problems = []
                for problem in temp_problems:
                    if "ConformanceException" in problem.problem and any(field_name in problem.problem for field_name in ['PercentMax', 'PercentMin', 'FixedMax', 'FixedMin', 'MfgCode']):
                        LOGGER.info(f"Ignoring complex conformance in global data type {name} from {f.name}: {problem.problem}")
                        continue
                    filtered_problems.append(problem)

                problems.extend(filtered_problems)

                # Global data types have no cluster IDs - they're truly global
                global_data_types[category][name] = XmlDataType(
                    data_type=data_type_enum,
                    name=name,
                    components=components,
                    cluster_ids=None  # Global types have no specific cluster IDs
                )

    # For now we assume we should have at least 3 global data type XMLs to parse
    # Intent here is to make user aware of typos in paths instead of silently having empty parsing
    if found_xmls < 3:
        raise SpecParsingException(f'Did not find all 3 global data model files in specified directory {top:!r}')

    return global_data_types, problems


def dm_from_spec_version(specification_version: uint) -> PrebuiltDataModelDirectory:
    ''' Returns the data model directory for a given specification revision.

        input: specification revision attribute data from the basic information cluster
        output: PrebuiltDataModelDirectory
        raises: ConformanceException if the given specification_version does not conform to a known data model
    '''
    # Specification version attribute is 2 bytes major, 2 bytes minor, 2 bytes dot 2 bytes reserved.
    # However, 1.3 allowed the dot to be any value
    if specification_version < uint(0x01040000):
        # The expression (specification_version & uint(0xFFFF00FF)) might be inferred as int by mypy.
        specification_version = typing.cast(uint, specification_version & uint(0xFFFF00FF))

    version_to_dm = {
        0x01030000: PrebuiltDataModelDirectory.k1_3,
        0x01040000: PrebuiltDataModelDirectory.k1_4,
        0x01040100: PrebuiltDataModelDirectory.k1_4_1,
        0x01040200: PrebuiltDataModelDirectory.k1_4_2,
        0x01050000: PrebuiltDataModelDirectory.k1_5,
    }

    if specification_version not in version_to_dm:
        raise ConformanceException(f"Unknown specification_version 0x{specification_version:08X}")

    return version_to_dm[specification_version]
